package com.greenline.push.service.core.utils.xmpp;

import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.jivesoftware.smack.ConnectionConfiguration;
import org.jivesoftware.smack.ConnectionConfiguration.SecurityMode;
import org.jivesoftware.smack.XMPPConnection;
import org.jivesoftware.smack.XMPPException;
import org.jivesoftware.smack.tcp.XMPPTCPConnection;
import org.jivesoftware.smackx.pubsub.LeafNode;
import org.jivesoftware.smackx.pubsub.PayloadItem;
import org.jivesoftware.smackx.pubsub.PubSubManager;
import org.jivesoftware.smackx.pubsub.SimplePayload;

import com.alibaba.fastjson.JSON;
import com.greenline.push.service.core.utils.Constants;

/**
 * @Type XmppManager
 * @Desc xmpp服务
 * @author kelei.huang
 * @date 2014年4月12日
 * @Version V1.0
 */
public class XmppManagerImpl implements XmppManager {

    Logger log = Constants.LOG_PROCESS;

    /**
     * xmpp地址
     */
    private String host;

    /**
     * xmpp端口
     */
    private Integer port;

    /**
     * xmpp域名
     */
    private String domain;

    /**
     * 管理员用户名
     */
    private String adminUserName;

    /**
     * 管理员密码
     */
    private String adminPassword;

    private XMPPConnection connection;

    private static Object lock = new Object();

    private static LinkedBlockingQueue<MyMessage> messageQueue = new LinkedBlockingQueue<MyMessage>(500);

    /**
     * 初始化
     */
    public void init() {
        for (int i = 0; i < 3; i++) {
            Thread t = new Thread(new sendMessageThread());
            t.start();
        }

    }

    /**
     * 获得xmpp连接
     * 
     * @return
     * @throws XMPPException
     */
    public XMPPConnection getXMPPConnection() {
        if (connection != null && !connection.isConnected()) {
            try {
                synchronized (lock) {
                    log.info("连接为空或者已经关闭.");
                    if (connection != null && !connection.isConnected()) {
                        connection.disconnect();
                        connection = null;
                    }
                }
            } catch (Exception e) {
                connection = null;
                log.error(e.getMessage(), e);
            }
        }
        if (connection == null) {

            synchronized (lock) {
                if (connection == null) {
                    log.info("连接开始.");
                    ConnectionConfiguration config = new ConnectionConfiguration(host, port, domain);
                    config.setSecurityMode(SecurityMode.disabled);
                    connection = new XMPPTCPConnection(config);
                    try {
                        connection.connect();
                        connection.login(adminUserName, adminPassword);
                        if (!connection.isAuthenticated()) {
                            log.error("登陆失败");
                            connection = null;
                        } else {
                            log.info("登陆成功.");
                        }
                    } catch (Exception e) {
                        log.error(e.getMessage(), e);
                        connection = null;
                        try {
                            // retry
                            TimeUnit.SECONDS.sleep(3l);
                        } catch (InterruptedException e1) {
                            e1.printStackTrace();
                            log.error(e1.getMessage());
                        }
                    }

                }
            }
        }
        return connection;
    }

    /**
     * 发布
     * 
     * @param nodeId
     * @param message
     */
    @Override
    public void publish(String nodeId, String context) {
        if (getXMPPConnection() == null) {
            return;
        }
        try {
            PubSubManager manager = new PubSubManager(getXMPPConnection());
            LeafNode myNode = null;
            try {
                myNode = manager.getNode(nodeId);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
            if (myNode == null) {
                myNode = manager.createNode(nodeId);
            }
            String namespace = "pubsub:" + nodeId;
            SimplePayload payload = new SimplePayload("message", namespace, context);
            PayloadItem<SimplePayload> item = new PayloadItem<SimplePayload>("5", payload);
            myNode.publish(item);
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 发送消息
     * 
     * @param userId
     * @param message
     * @throws Exception
     */
    @Override
    public void send(String userId, MessageVO message) throws Exception {
        String context = JSON.toJSONString(message);
        send(userId, context);
    }

    /**
     * 发送消息
     * 
     * @param userId
     * @param context
     * @throws Exception
     */
    @Override
    public void send(String userId, String context) throws Exception {
        MyMessage msg = new MyMessage(userId + "@" + domain, MyMessage.Type.push);
        msg.setBody(context);
        messageQueue.put(msg);
    }

    /**
     * 发送消息
     */
    public void sendMessage() {
        while (true) {
            try {
                XMPPConnection conn = getXMPPConnection();
                if (conn != null) {
                    MyMessage message = messageQueue.take();
                    log.info("发送信息" + JSON.toJSONString(message));
                    conn.sendPacket(message);
                }
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        }
    }

    public String getHost() {
        return host;
    }

    public void setHost(String host) {
        this.host = host;
    }

    public Integer getPort() {
        return port;
    }

    public void setPort(Integer port) {
        this.port = port;
    }

    public String getDomain() {
        return domain;
    }

    public void setDomain(String domain) {
        this.domain = domain;
    }

    public String getAdminUserName() {
        return adminUserName;
    }

    public void setAdminUserName(String adminUserName) {
        this.adminUserName = adminUserName;
    }

    public String getAdminPassword() {
        return adminPassword;
    }

    public void setAdminPassword(String adminPassword) {
        this.adminPassword = adminPassword;
    }

    class sendMessageThread implements Runnable {

        @Override
        public void run() {
            while (true) {
                try {
                    XMPPConnection conn = getXMPPConnection();
                    if (conn != null) {
                        MyMessage message = messageQueue.take();
                        log.info("发送信息" + JSON.toJSON(message));
                        conn.sendPacket(message);
                    }
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
            }
        }
    }
}
